package com.peaksport.pkms.leanrtest2.service;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class TestKafka2Receiver {
	

	@KafkaListener(topics = "peak002")
	public void listen(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws Exception {
	    //判断是否NULL
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
 
        if (kafkaMessage.isPresent()) {
            //获取消息
         //   Object message = kafkaMessage.get();
		    log.info("Kafka Message Receiver2:" + record.toString());
		    Thread.sleep(300);
        }
	}
	
	
//	@KafkaListener(id = "tuta2", topics = "pkkafkatut")
//	public void listen2(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
//	    //判断是否NULL
//        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
// 
//        if (kafkaMessage.isPresent()) {
//            //获取消息
//        	
//            Object message = kafkaMessage.get();
//		    log.info("Receive： +++++++++++++++ Message:" + message);
//        }
//        
//	}
}
